from typing import Any, Dict, List, Optional, Callable
import json
import time
from urllib import request as urlreq
import asyncio
import os

from .base_node import BaseNode


class StartNode(BaseNode):
    """
    起始节点：支持多种触发模式，生成初始输出并进入默认后继。
    触发模式（data.mode）：
    - manual（默认）：从 context.globals.input 或顶层 input/seed 取初始输出
    - webhook：从 context.trigger.payload 取初始输出（用于外部 webhook 调用场景）
    - http：发起 HTTP 请求（data.url/method/headers/timeout），以响应内容作为初始输出
    - kafka：监听 Kafka（data.kafkaBinding 或 kafkaTopic+kafkaGroup+kafkaTimeout），取首条匹配消息作为初始输出
    - file：读取本地文件（data.path，data.format=json|text），作为初始输出
    - schedule/interval：记录触发元数据（firedAt/interval），payload 从 globals.input 回退
    输出结构约定：{"payload": Any, "trigger": {"type": str, "meta": {...}}}
    """
    type = 'event'

    def execute(self, graph: 'FlowGraph', context: Dict[str, Any], log_cb: Optional[Callable[[str], None]] = None) -> List[str]:
        logger = log_cb or (lambda m: None)
        data = self.data or {}
        mode = str(data.get('mode') or 'manual').strip().lower()
        output: Dict[str, Any] = {"payload": {}, "trigger": {"type": mode, "meta": {}}}

        def _from_globals_input() -> Dict[str, Any]:
            try:
                gi = context.get('globals', {}).get('input') if isinstance(context.get('globals'), dict) else None
                if gi is None:
                    gi = context.get('input') or context.get('seed')
                if isinstance(gi, dict):
                    return gi
            except Exception:
                pass
            return {}

        try:
            if mode == 'manual':
                payload = _from_globals_input()
                output["payload"] = payload
            elif mode == 'webhook':
                payload = {}
                try:
                    payload = (context.get('trigger') or {}).get('payload') or {}
                except Exception:
                    payload = {}
                output["payload"] = payload
            elif mode == 'http':
                url = data.get('url')
                method = (data.get('method') or 'GET').upper()
                headers = data.get('headers') or {"Content-Type": "application/json"}
                timeout = int(data.get('timeout') or 30)
                if not url:
                    logger(f"[StartNode] http mode missing url; fallback to globals.input")
                    output["payload"] = _from_globals_input()
                else:
                    req = urlreq.Request(url=url, method=method, headers=headers)
                    with urlreq.urlopen(req, timeout=timeout) as resp:
                        body_text = resp.read().decode('utf-8', errors='replace')
                        try:
                            payload = json.loads(body_text)
                        except Exception:
                            payload = {"raw": body_text}
                        output["payload"] = payload
                        output["trigger"]["meta"]["statusCode"] = resp.getcode()
            elif mode == 'kafka':
                kafka_binding = data.get('kafkaBinding')
                kafka_topic = data.get('kafkaTopic')
                kafka_group = data.get('kafkaGroup')
                kafka_timeout = float(data.get('kafkaTimeout') or 5)
                kafka_expect_key = str(data.get('kafkaExpectKey') or '').strip()
                expect_val = None
                try:
                    expect_val = str((context.get('flow') or {}).get('runId') or '').strip()
                except Exception:
                    expect_val = None
                payload_box: Dict[str, Any] = {"_received": None}

                async def handler(msg: Any):
                    try:
                        if kafka_expect_key and isinstance(msg, dict):
                            val = str(msg.get(kafka_expect_key) or '').strip()
                            if expect_val and val != expect_val:
                                return
                        payload_box["_received"] = msg
                    except Exception:
                        payload_box["_received"] = msg

                try:
                    from api.managers.kafka_manager import KAFKA_MANAGER
                    if kafka_binding:
                        task_factory = KAFKA_MANAGER.consumer_task_by_binding(kafka_binding, handler)
                    else:
                        task_factory = KAFKA_MANAGER.create_consumer_task([kafka_topic], kafka_group, handler, decode_json=True)
                    async def _run_until(timeout_sec: float):
                        t = asyncio.create_task(task_factory())
                        start = time.time()
                        while time.time() - start < timeout_sec and payload_box.get('_received') is None:
                            await asyncio.sleep(0.2)
                        return t
                    asyncio.run(_run_until(kafka_timeout))
                except Exception as e:
                    logger(f"[StartNode] kafka listen error: {e}")
                output["payload"] = payload_box.get('_received') or {}
            elif mode == 'file':
                path = data.get('path')
                fmt = str(data.get('format') or 'json').lower()
                try:
                    if path and os.path.exists(path):
                        with open(path, 'r', encoding='utf-8') as f:
                            text = f.read()
                            if fmt == 'json':
                                try:
                                    payload = json.loads(text)
                                except Exception:
                                    payload = {"raw": text}
                            else:
                                payload = {"raw": text}
                            output["payload"] = payload
                    else:
                        output["payload"] = _from_globals_input()
                except Exception as e:
                    logger(f"[StartNode] file read error: {e}")
                    output["payload"] = _from_globals_input()
            elif mode == 'event' or (str(data.get('triggerType') or '').lower() == 'event'):
                payload = {}
                try:
                    payload = (context.get('event') or {})
                except Exception:
                    payload = {}
                output["payload"] = payload
        except Exception as e:
            logger(f"[StartNode] trigger execute error: {e}")
            output["payload"] = _from_globals_input()
        self.output = output
        # 兼容旧行为：将 payload 的键提升到顶层，保留新结构（payload/trigger）
        try:
            if isinstance(output.get('payload'), dict):
                compat = dict(output)
                compat.update(output['payload'])
                self.output = compat
        except Exception:
            pass
        logger(f"[StartNode] Trigger={mode} payload keys={list((output or {}).get('payload', {}).keys())}")
        return [e.target for e in graph.get_outgoing_edges(self.id) if e.source_handle in (None, 'default')]